/*
 * Hyuga: A Density-Grid Stream Clustering Platform.
 * Copyright (C) 2014 PUC-Rio/Laboratory for Advanced Collaboration
 *
 * Hyuga is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Hyuga is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Hyuga. If not, see <http://www.gnu.org/licenses/>.
 */
package br.pucrio.inf.lac.konoha.hyuga.others.csv;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

import org.apache.log4j.Logger;

import com.espertech.esper.adapter.InputAdapter;
import com.espertech.esperio.AdapterInputSource;
import com.espertech.esperio.csv.CSVInputAdapter;
import com.espertech.esperio.csv.CSVInputAdapterSpec;

import br.pucrio.inf.lac.konoha.hyuga.core.Bootstrap;

/**
 * A CSV playback program. 
 * 
 * This program uses the csvplayback.properties to configure a Hyuga EPA to replay the content of the input csv file.
 * The input csv file must have the following columns:
 * uuid,latitude,longitude,timestamp
 * 
 * Sample File:
 * uuid, latitude,          longitude,        timestamp
 * 286,  -22.9559986550780, -43.1922377359216, 70462
 * 973,  -22.9603834270102, -43.2000845051359, 70462
 * 2742, -22.9728237658448, -43.1907249148011, 704637
 * 4075, -22.9788327380116, -43.1908675079309, 70464
 * 878,  -22.9811741072085, -43.2333375518404, 70464
 * 3431, -22.9703160507337, -43.1907072643957, 704650
 * 341,  -22.9656428061742, -43.2163599814267, 70466
 * 5028, -22.9580526401979, -43.2050608584764, 704664
 *
 * @author Marcos Roriz <mroriz@inf.puc-rio.br>
 */
public class CSVPlayback {

    /** LOGGER. */
    private static Logger LOGGER = Logger.getLogger("br.pucrio.inf.lac.konoha.hyuga");
    
    /** Hyuga. */
    public Bootstrap bootSystem;
    
    /** Scenario Properties. */
    public Properties scenario;
    
    /**
     * Entry point for the main routine.
     * 
     * @param args the scenario and csv playback files.
     */
    public static void main(String[] args) {
        String scenarioFileName = "";
        String datasetFileName  = "";

        // Reading command line arguments
        if (args.length == 0 || args.length > 2) {
            LOGGER.error("Invalid number of arguments");
            LOGGER.error("Usage: java -jar csvplayback.jar [scenario.properties] dataset.csv");
            System.exit(-1);
        } else if (args.length == 1) { 
            scenarioFileName = "csvplayback.properties";
            datasetFileName  = args[0];
        } else {
            scenarioFileName = args[0];
            datasetFileName  = args[1];
        }
        
        // Building        
        CSVPlayback csvPlayback = new CSVPlayback();
        csvPlayback.loadScenario(scenarioFileName);
        csvPlayback.boot();

        // Starting Playback
        LOGGER.info("CSV Playback...");
        csvPlayback.play(datasetFileName);
                
        System.exit(0);
    }

    public void loadScenario(String filename) {
        // Loading application parameters
        scenario = null;
        try {
            LOGGER.info("Loading parameters from " + filename);
            scenario = readPropertyFile(filename);
        } catch (IOException ex) {
            LOGGER.error("Error at reading the configuration file");
            System.exit(-1);
        }
    }
    
    public void boot() {
        // Booting the system!
        bootSystem = new Bootstrap(scenario);
        
        try {
            bootSystem.buildScenario();
            bootSystem.buildDataStream();
            bootSystem.deployHyuga();
        } catch (IllegalArgumentException ex) {
            LOGGER.error(ex.getMessage());
            System.exit(-1);
        }
    }
    
    public void play(String fileName) {
        // Loading the File
        File file = new File(fileName);
        if (!file.exists()) {
            LOGGER.error("Error at reading the dataset file: " + fileName);
            System.exit(-1);
        }
        
        System.out.println("Sleeping so we can update the DDS topics...");
        try {
            Thread.sleep(30000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // Creating the Playback
        AdapterInputSource source = new AdapterInputSource(file);
        CSVInputAdapterSpec spec  = new CSVInputAdapterSpec(source, "CSVLocationUpdateEvent");
        
        spec.setTimestampColumn("timestamp");
        InputAdapter inputAdapter = new CSVInputAdapter(bootSystem.getHyuga().getCEPEngine(), spec);

        // Play
        LOGGER.info("Replay Started...");
        long before = System.currentTimeMillis();
        bootSystem.getDataStream().publish("Meta", "RESET", "COMMAND");
        bootSystem.getDataStream().publish("Manager", "RESET", "COMMAND");
        inputAdapter.start();
        long after = System.currentTimeMillis();

        // Exiting
        long diff = after - before;
        LOGGER.info("Replay took " + (diff) + " milliseconds");
        LOGGER.info("Replay took " + (diff) / 1000 + " seconds");
        LOGGER.info("Replay took " + (diff) / 1000 / 60 + " minutes");
    }

    private static Properties readPropertyFile(String filename) throws IOException {
        Properties properties = new Properties();

        // Try to load indicated property file
        File propertyFile = new File(filename);

        if (!propertyFile.exists()) {
            LOGGER.warn("File " + filename + " not found in the current directory");
            LOGGER.warn("Trying to load " + filename + " using the classpath");

            InputStream is = Thread.currentThread().getContextClassLoader().getResourceAsStream(filename);
            if (is != null) {
                properties.load(is);
                LOGGER.info("Loaded parameters from file: " + filename);
            } else {
                LOGGER.error("File " + filename + " does not exist");
                throw new FileNotFoundException();
            }
        } else {
            FileInputStream fi = new FileInputStream(propertyFile);
            LOGGER.info("Loaded parameters from file: " + propertyFile.getName());
            properties.load(fi);
            fi.close();
        }
        return properties;
    }

}
